/*
 * 文件名：RealEventCounter.java
 * 版权：亚信联创版权所有
 * 描述：实时流日志数据监控类
 * 修改人：xuwei3
 * 修改时间：2015-5-13
 * 修改内容：
 */
package com.ailk.bigdata.etl.realstream.server.monitor;

import java.io.BufferedInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.sql.Timestamp;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.PropertyResourceBundle;
import java.util.ResourceBundle;
import java.util.StringTokenizer;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ailk.bigdata.etl.common.server.database.JdbcUtil;
import com.ailk.bigdata.etl.common.server.date.DateUtil;
import com.ailk.bigdata.etl.common.server.monitor.MonitoredCounterGroup;
import com.ailk.bigdata.etl.common.server.thread.ThreadPool;
import com.ailk.bigdata.etl.realstream.server.tools.RealStreamTool;

/**
 * @description 实时流日志数据监控类
 * @author [xuwei3]
 * @version [版本号,2015-5-13]
 * @see [相关类/方法]
 * @since [产品/模块版本]
 */
public class RealEventCounter extends MonitoredCounterGroup implements EventCounter
{
	private static Logger logger = LoggerFactory.getLogger(RealEventCounter.class);

	private static final String COUNTER_EVENTS_RECEIVED = "events.received";
	private static final String COUNTER_DATAS_ACCEPTED = "datas.accepted";
	private static final String COUNTER_EVENTS_SENDED = "events.sended";

	public static final String[] DEFAULT_ATTRIBUTES = { COUNTER_EVENTS_RECEIVED, COUNTER_DATAS_ACCEPTED, COUNTER_EVENTS_SENDED };

	public RealEventCounter(String name)
	{
		super(MonitoredCounterGroup.Type.SOURCE, name, DEFAULT_ATTRIBUTES);
	}

	public RealEventCounter(String name, String[] attributes)
	{
		super(MonitoredCounterGroup.Type.SOURCE, name, attributes);
	}

	@Override
	public boolean reSetCounter()
	{
		boolean returnVal = isChanged;
		stopTime.set(0L);
		for (String counter : counterMap.keySet())
		{
			counterMap.get(counter).set(0L);
			logger.debug("统计类型: " + type + ", name: " + name + ", attribute: " + counter + " recounted");
		}
		startTime.set(System.currentTimeMillis());
		isChanged = false;
		logger.debug("统计类型: " + type + ", name: " + name + " started");
		return returnVal;
	}

	/**
	 * 
	 * @description 获取统计数据服务
	 * @param name
	 * @param attributes
	 * @return
	 */
	public static RealEventCounter getRealEventCounter(String name, List<String> attribute)
	{
		attribute.addAll(Arrays.asList(DEFAULT_ATTRIBUTES));
		String[] attributes = new String[attribute.size()];
		for (int i = 0; i < attribute.size(); i++)
		{
			attributes[i] = attribute.get(i);
		}
		return new RealEventCounter(name, attributes);
	}

	public long getEventReceivedCountByType(String interType)
	{
		return get(interType);
	}

	public long incrementEventReceivedByType(String interType)
	{
		return increment(interType);
	}

	public long getEventSendedCountByType(String interType)
	{
		return get(interType);
	}

	public long incrementEventSendedByType(String interType)
	{
		return increment(interType);
	}

	public long getEventReceivedCount()
	{
		return get(COUNTER_EVENTS_RECEIVED);
	}

	public long incrementEventReceivedCount()
	{
		return increment(COUNTER_EVENTS_RECEIVED);
	}

	public long addToEventReceivedCount(long delta)
	{
		return addAndGet(COUNTER_EVENTS_RECEIVED, delta);
	}

	public long getDatasAcceptedCount()
	{
		return get(COUNTER_DATAS_ACCEPTED);
	}

	public long incrementDatasAcceptedCount()
	{
		return increment(COUNTER_DATAS_ACCEPTED);
	}

	public long addToDatasAcceptedCount(long delta)
	{
		return addAndGet(COUNTER_DATAS_ACCEPTED, delta);
	}

	public long getEventSendedCount()
	{
		return get(COUNTER_EVENTS_SENDED);
	}

	public long incrementEventSendedCount()
	{
		return increment(COUNTER_EVENTS_SENDED);
	}

	public long addToEventSendedCount(long delta)
	{
		return addAndGet(COUNTER_EVENTS_SENDED, delta);
	}

	public void startSchedulerdMonitor()
	{
		ThreadPool.getScheduleThreadPool("log-monitor-agent", 1).scheduleAtFixedRate(new LogMonitorTask(this), 1, 60, TimeUnit.SECONDS);
		logger.info("agent日志管理线程启动，每2分钟执行一次，首次执行在[" + 3600 + "]millis后启动.........");
	}

	public long GetAndResetCount(String counter)
	{
		long current = counterMap.get(counter).get();
		counterMap.get(counter).set(0L);
		return current;
	}
}

/**
 * 
 * @description
 * @author [xuwei3]
 * @version [版本号,2015-7-14]
 * @see [相关类/方法]
 * @since [产品/模块版本]
 */
class LogMonitorTask implements Runnable
{
	private static Logger logger = LoggerFactory.getLogger(LogMonitorTask.class);

	private static Map<String, String> subscribeInfos = new HashMap<String, String>();

	private String INSERT_SMS_SQL = "INSERT INTO td_send_sms(phone_num, message_content, send_status, create_time) VALUES (?, ?, ?, ?)";
	private String content = "该时间段内没有接受到数据,请检查相关配置,当前时间是%s";
	private String fileSeparator = System.getProperty("file.separator");
	private String currentDir = System.getProperty("user.dir");// 当前路径
	private String currentParentDir = currentDir.substring(0, currentDir.lastIndexOf(fileSeparator));// bin的上级目录
	private String confPath = currentParentDir + fileSeparator + "conf" + fileSeparator + "config.properties";
	private long loatModfiedTime = System.currentTimeMillis();

	private RealEventCounter realEventCounter;

	/**
	 * @param realEventCounter
	 */
	public LogMonitorTask(RealEventCounter realEventCounter)
	{
		this.realEventCounter = realEventCounter;
	}

	/**
	 * @see java.lang.Runnable#run()
	 */
	@Override
	public void run()
	{
		logger.info("实时流监控日志:" + realEventCounter.toString());
		if (!realEventCounter.reSetCounter())
		{
			try
			{
				File conFile = new File(confPath);
				if (conFile.exists() && conFile.isFile())
				{
					if (conFile.lastModified() != loatModfiedTime)
					{
						subscribeInfos.clear();
						BufferedInputStream inputStream = new BufferedInputStream(new FileInputStream(conFile));
						ResourceBundle bundle = new PropertyResourceBundle(inputStream);
						String subscribe = bundle.getString("SUBSCRIBER");
						if (StringUtils.isNotEmpty(subscribe))
						{
							StringTokenizer token = new StringTokenizer(subscribe, ";");
							while (token.hasMoreTokens())
							{
								String nextToken = token.nextToken();
								String[] subInfo = StringUtils.split(nextToken, ":");
								logger.info("信息订阅人信息[" + nextToken + "]");
								subscribeInfos.put(subInfo[0], subInfo[1]);
							}
						}
					}
				}
				
				if (!subscribeInfos.isEmpty())
				{
					logger.warn("该时间段内没有接受到数据,发送短信告警.........");

					Object[][] parms = new Object[subscribeInfos.size()][4];
					int flag = 0;
					for (String phone : subscribeInfos.values())
					{
						parms[flag][0] = phone;
						parms[flag][1] = String.format(content, DateUtil.getToday("yyyyMMddHHmm"));
						parms[flag][2] = 0;
						parms[flag++][3] = new Timestamp(System.currentTimeMillis());
					}
					JdbcUtil.batchExecute(JdbcUtil.getConnection(RealStreamTool.dbInfo), INSERT_SMS_SQL, parms, true);
					parms = null;
				}
			}
			catch (Exception e)
			{
				logger.warn("发送告警短信异常", e);
			}
		}
	}
}
